package bx.job.actors;

import akka.actor.typed.ActorRef;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Receive;
import bx.job.JobServerBootstrap;
import bx.job.entity.BxTriggerInfo;
import bx.job.jobstore.bean.JobInfoContextBean;
import bx.job.lock.bean.ScheduleUnlockContextBean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CompletableFuture;

public class JobProcessManagerActor extends AbstractBehavior<JobProcessManagerActor.JobProcessManagerActorMsg> {
    private final static Logger log = LoggerFactory.getLogger(JobProcessManagerActor.class);
    private ActorRef<LockManagerActor.LockManagerActorMsg> lockManagerActorActor;
    private ActorRef<JobIssueActor.JobIssueActorMsg> jobIssueActor;
    private ActorRef<JobLogActor.JobIssueActorMsg> jobLogActor;
    private int unlockCounter;

    public static Behavior<JobProcessManagerActor.JobProcessManagerActorMsg> create(
            ActorRef<LockManagerActor.LockManagerActorMsg> lockManagerActorActor,
            ActorRef<JobLogActor.JobIssueActorMsg> jobLogActor, ActorRef<JobIssueActor.JobIssueActorMsg> jobIssueActor) {
        return Behaviors.setup(context -> new JobProcessManagerActor(context, lockManagerActorActor, jobLogActor, jobIssueActor));
    }

    public JobProcessManagerActor(ActorContext<JobProcessManagerActorMsg> context,
                                  ActorRef<LockManagerActor.LockManagerActorMsg> lockManagerActorActor,
                                  ActorRef<JobLogActor.JobIssueActorMsg> jobLogActor,
                                  ActorRef<JobIssueActor.JobIssueActorMsg> jobIssueActor) {
        super(context);
        this.lockManagerActorActor = lockManagerActorActor;
        this.jobLogActor = jobLogActor;
        this.jobIssueActor = jobIssueActor;
    }

    @Override
    public Receive<JobProcessManagerActorMsg> createReceive() {
        return newReceiveBuilder().onMessage(JobProcessManagerActor.IssueJobFromTriggerMsg.class, jobFromTriggerMsg -> {
            BxTriggerInfo bxTriggerInfo = jobFromTriggerMsg.triggerInfo();
            CompletableFuture.supplyAsync(() -> JobServerBootstrap.getJobStore().getJobList(bxTriggerInfo.getTriggerId())).thenAccept(jobList -> {
                this.unlockCounter += jobList.size();
                jobList.forEach(jobInfo -> {
                    JobInfoContextBean jobInfoContext = JobInfoContextBean.builder().triggerInfo(bxTriggerInfo).bxJobInfo(jobInfo).build();
                    jobLogActor.tell(new JobLogActor.JobIssueMsg(jobInfoContext, this.getContext().getSelf()));
                });
            });
            return this;
        }).onMessage(JobProcessManagerActor.JobLoggedMsg.class, jobLoggedMsg -> {
            jobIssueActor.tell(new JobIssueActor.JobIssueMsg(jobLoggedMsg.jobInfoContextBean()));
            unlockCounter--;
            if (unlockCounter < 0) {
                log.error("error unlockCounter ???");
                // TODO 解决这种异常情况，但实际上这不应该出现
            }
            if (unlockCounter == 0) {
                ScheduleUnlockContextBean unLockContext = ScheduleUnlockContextBean.builder().build();
                lockManagerActorActor.tell(new LockManagerActor.UnlockMsg(unLockContext, this.getContext().getSelf()));
                return this;
            }
            return this;
        }).onMessage(JobProcessManagerActor.UnlockedMsg.class, unlockedMsg -> Behaviors.stopped()).build();
    }

    public interface JobProcessManagerActorMsg {
    }


    public static class UnlockedMsg implements JobProcessManagerActorMsg {
    }


    public static record IssueJobFromTriggerMsg(BxTriggerInfo triggerInfo) implements JobProcessManagerActorMsg {
    }

    public static record JobLoggedMsg(JobInfoContextBean jobInfoContextBean) implements JobProcessManagerActorMsg {
    }

}
